package com.lzy.canal.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.lzy.canal.component.consumer.mysql.DefaultDdlConsumer;
import com.lzy.canal.component.consumer.mysql.DefaultDmlConsumer;
import com.lzy.canal.component.filter.mysql.MySqlByConfigFilter;
import com.lzy.canal.register.consumer.ConsumerRegisterManage;
import com.lzy.canal.register.consumer.DdlConsumerRegister;
import com.lzy.canal.register.consumer.DmlConsumerRegister;
import com.lzy.canal.register.consumer.MysqlFilterRegister;
import com.lzy.canal.service.CanalConsumerService;
import com.lzy.canal.service.CanalMessageService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Canal配置
 *
 * @author lzy
 */
@Configuration
@EnableConfigurationProperties({CanalClientProperty.class})
@ConditionalOnProperty(name = "com.lzy.canal.client.property.enabled", havingValue = "true")
public class CanalClientAutoConfig {
    @Autowired
    private CanalClientProperty canalProperty;
    @Autowired
    private ConsumerRegisterManage registerManage;

    /**
     * 连接管理器
     *
     * @return
     */
    @Bean
    @ConditionalOnMissingBean
    public CanalClientConnectorManage getCanalConnector() {
        Map<String, CanalConnector> map = canalProperty.getProperItems().stream()
                .collect(Collectors.toMap(CanalClientProperty.CanalProperty::getDestination, this::createCanalConnector));
        return new CanalClientConnectorManage(map);
    }

    /**
     * table 过滤器
     *
     * @return
     */
    @Bean
    @ConditionalOnMissingBean
    public MySqlByConfigFilter getMySqlFilterByConfig() {
        return new MySqlByConfigFilter(canalProperty);
    }



    @Bean
    @ConditionalOnMissingBean
    public CanalMessageService getCanalMessageService(CanalConsumerService canalConsumerService, MysqlFilterRegister filterRegister) {
        return new CanalMessageService(canalConsumerService, filterRegister);
    }

    @Bean
    @ConditionalOnMissingBean
    public CanalConsumerService getCanalConsumerService() {
        return new CanalConsumerService(registerManage);
    }


    /**
     * canal client 连接管理器
     */
    public static class CanalClientConnectorManage {
        private final Map<String, CanalConnector> map;

        public CanalClientConnectorManage(Map<String, CanalConnector> map) {
            this.map = map;
        }

        /**
         * 获取 CanalConnector
         *
         * @param destination destination
         * @return
         */
        public CanalConnector get(String destination) {
            return map.get(destination);
        }
    }

    /**
     * 创建canal 客户端
     *
     * @param item 属性配置
     * @return
     */
    private CanalConnector createCanalConnector(CanalClientProperty.CanalProperty item) {
        return CanalConnectors.newClusterConnector(Collections.singletonList(new InetSocketAddress(item.getHostname(), item.getPort())),
                item.getDestination(),
                item.getUsername(),
                item.getPassword());
    }


    /**
     * @author lzy
     */
    @Configuration
    @ConditionalOnProperty(name = "com.lzy.canal.client.config.enabled", havingValue = "true")
    public static class CanalConsumerAutoConfiguration {

        @Bean
        @ConditionalOnMissingBean
        public DefaultDdlConsumer getDefaultDdlConsumer() {
            return new DefaultDdlConsumer();
        }

        @Bean
        @ConditionalOnMissingBean
        public DefaultDmlConsumer getDefaultDmlConsumer() {
            return new DefaultDmlConsumer();
        }

        @Bean
        @ConditionalOnMissingBean
        public DdlConsumerRegister getDdlConsumerRegister(DefaultDdlConsumer consumer) {
            return new DdlConsumerRegister(consumer);
        }

        @Bean
        @ConditionalOnMissingBean
        public DmlConsumerRegister getDmlInsertOrUpdateConsumerRegister(DefaultDmlConsumer consumer) {
            return new DmlConsumerRegister(consumer);
        }

        @Bean
        @ConditionalOnMissingBean
        public ConsumerRegisterManage getConsumerRegisterManage(DdlConsumerRegister ddlConsumerRegister, DmlConsumerRegister dmlConsumerRegister) {
            return new ConsumerRegisterManage(ddlConsumerRegister, dmlConsumerRegister);
        }

        @Bean
        @ConditionalOnMissingBean
        public MysqlFilterRegister getMysqlFilterRegister(MySqlByConfigFilter filter) {
            return new MysqlFilterRegister(filter);
        }

    }
}
